package fda.schema;

import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;

import fda.base.Tuple;

public class TableInfo {

	private final Schema parent;
	private String dataPath;
	private int primeKeyIndex = -1;
	private final List<Column> cols = new LinkedList<>();
	private int tupleLength;

	public TableInfo(String dataPath, Schema parent) {
		this.tupleLength = 0;
		this.parent = parent;
		this.dataPath = dataPath;
	}

	public Tuple joinTuples(Tuple t1, Tuple t2) {

		if (t1.getKeyValue() != t2.getKeyValue())
			throw new IllegalStateException(String.format("Cannot Join Tuples with different keys: %d adn %d",
					t1.getKeyValue(), t2.getKeyValue()));

		// read all values from both columns
		HashMap<String, String> colVal = new HashMap<>();
		readColValuePairs(t1, colVal);
		readColValuePairs(t2, colVal);

		// create a line of text for merged tuple
		StringBuilder sb = new StringBuilder();
		sb.append(String.format("%-" + (t1.getKeyLength()) + "s", t1.getKeyValue()));
		String keyColName = cols.get(primeKeyIndex).name;
		for (Column c : cols) {
			// skip first Key column (it is already added)
			if (c.name.equalsIgnoreCase(keyColName))
				continue;
			sb.append(String.format("%" + (c.length) + "s", colVal.get(c.name)));
		}

		return new Tuple(sb.toString(), this);
	}

	/**
	 * Read all columnName-Value pairs into out_colVal
	 * @param tuple tuple to read from
	 * @param out_colVal
	 */
	private void readColValuePairs(Tuple tuple, HashMap<String, String> out_colVal) {
		TableInfo ti1 = tuple.getTableInfo();
		String keyColName = tuple.getKeyColumnName();
		for (Column c : ti1.getCols()) {
			if (keyColName.equalsIgnoreCase(c.name))
				continue;
			out_colVal.put(c.name, tuple.getValue(c.name));
		}
	}

	/**
	 * Copy-constructor to create a copy of source table with different location (dataPath)
	 * @param source TableInfo to copy from 
	 * @param dataPath place where data of this file remains
	 */
	public TableInfo(TableInfo source, String dataPath) {
		this.tupleLength = source.tupleLength;
		this.parent = source.parent;
		this.primeKeyIndex = source.primeKeyIndex;
		for (Column col : source.cols)
			this.cols.add(col);
		this.tupleLength = source.tupleLength;
		this.dataPath = dataPath;
	}

	public void addCol(Column column) {
		tupleLength += column.length;
		cols.add(column);
	}

	public void setDataPath(String dataPath) {
		this.dataPath = dataPath;
	}

	/**
	 * @return 0(zero) based index of primary key column (column on which join is executed)
	 */
	public int getPrimeKeyIndex() {
		return primeKeyIndex;
	}

	/**
	 * @param primeKeyIndex  0(zero) based index of primary key.
	 */
	public void setPrimeKeyIndex(int primeKeyIndex) {
		this.primeKeyIndex = primeKeyIndex;
	}

	public List<Column> getCols() {
		return cols;
	}

	public String getDataPath() {
		return dataPath;
	}

	public int getTupleLength() {
		return tupleLength;
	}

	@Override
	public String toString() {
		StringBuilder sb = new StringBuilder();
		String line = "-------------------------------------";

		sb.append("\n" + line);
		sb.append("\n").append(dataPath);
		for (int i = 0; i < cols.size(); ++i) {
			String info;
			Column c = cols.get(i);

			if (i == primeKeyIndex) {
				info = String.format("%s\t%d\tkey", c.name, c.length);
			}
			else {
				info = String.format("%s\t%d", c.name, c.length);
			}
			sb.append("\n").append(info);
		}
		sb.append("\n" + line);
		return sb.toString();
	}

}
